/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements.  See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership.  The ASF licenses this file to you under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with the License.  You may obtain
 * a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package org.apache.zookeeper;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
import java.util.Set;

import org.apache.zookeeper.ClientCnxn.EndOfStreamException;
import org.apache.zookeeper.ClientCnxn.Packet;
import org.apache.zookeeper.ZooDefs.OpCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ClientCnxnSocketNIO extends ClientCnxnSocket {
    private static final Logger LOG = LoggerFactory.getLogger(ClientCnxnSocketNIO.class);

    private final Selector selector = Selector.open();

    /**
     * {@link ClientCnxnSocketNIO#registerAndConnect}之后，sockKey才不为null.
     */
    private SelectionKey sockKey;

    ClientCnxnSocketNIO() throws IOException {
        super();
    }

    @Override
    boolean isConnected() {
        return sockKey != null;
    }

    /**
     * @return true if a packet was received
     * @throws InterruptedException
     * @throws IOException
     */
    void doIO(List<Packet> pendingQueue, LinkedList<Packet> outgoingQueue, ClientCnxn cnxn) throws InterruptedException, IOException {
        SocketChannel sock = (SocketChannel) sockKey.channel();
        if (sock == null) {
            throw new IOException("Socket is null!");
        }

        // 【socket可读】
        if (sockKey.isReadable()) {
            int rc = sock.read(incomingBuffer);
            if (rc < 0) {
                throw new EndOfStreamException("Unable to read additional data from server sessionid 0x" + Long.toHexString(sessionId) + ", likely server has closed socket");
            }
            // 如果上面sock.read(incomingBuffer)读完一段数据后，byteBuffer没有剩余空间了:
            if (!incomingBuffer.hasRemaining()) {
                // 被‘写满’的byteBuffer：flip之后，就可以从incomingBuffer‘读数据’了。
                incomingBuffer.flip();

                // 【1、】如果和 lenBuffer 相等的话，就说明incomingBuffer读取的是数据包前四个字节：即数据包长度len
                if (incomingBuffer == lenBuffer) {
                    // 接收数据包总数++
                    recvCount++;
                    // fixme-lxf:疑问：这里只读取了数据包总长度length，并且创建了一个length长度的byteBuffer；但没有把socket中数据读取到incomingBuffer中。这是什么原理的？
                    // 读取长度，并allocate一个len长度的byteBuffer
                    readLength();

                } else if (!initialized) {
                    // 【2、】如果客户端session尚未初始化，则说明是‘建立session连接的相应’。处理完响应后，设置 标识符initialized = true

                    // 读取connect请求的结果
                    readConnectResult();

                    // 开启读操作
                    enableRead();
                    // 从outgoingQueue中，如果寻找到‘可发送的’数据包，就开启写操作。
                    if (findSendablePacket(outgoingQueue, cnxn.sendThread.clientTunneledAuthenticationInProgress()) != null) {
                        // Since SASL authentication has completed (if client is configured to do so),
                        // outgoing packets waiting in the outgoingQueue can now be sent.
                        enableWrite();
                    }

                    lenBuffer.clear();
                    incomingBuffer = lenBuffer;
                    // 修改最后接收数据事件
                    updateLastHeard();
                    // 已初始化
                    initialized = true;

                } else {
                    // 【3、】其他情况，就是客户端发送的请求了。
                    sendThread.readResponse(incomingBuffer);

                    // 重置两个byteBuffer，等待下次请求过来，使用
                    lenBuffer.clear();
                    incomingBuffer = lenBuffer;

                    updateLastHeard();
                }
            }
        }

        // 【socket可写】
        if (sockKey.isWritable()) {
            synchronized (outgoingQueue) {
                // 当前是socket写事件：从outgoingQueue中找到一个‘可写的’packet，序列化之后 ，通过网络io发送出去。
                Packet p = findSendablePacket(outgoingQueue, cnxn.sendThread.clientTunneledAuthenticationInProgress());
                if (p != null) {
                    updateLastSend();
                    // 判断是否要把packet的数据写到byteBuffer中。即p.bb
                    if (p.bb == null) {
                        if ((p.requestHeader != null) && (p.requestHeader.getType() != OpCode.ping) && (p.requestHeader.getType() != OpCode.auth)) {
                            // 这里生成xid并设置到请求头中。【p.requestHeader.xid生成】
                            p.requestHeader.setXid(cnxn.getXid());
                        }
                        p.createBB();
                    }
                    // 【1、】使用sock把packet中的byteBuffer写到网络io中。
                    sock.write(p.bb);

                    // 判断p.bb这个byteBuffer中的所有数据，是否已经全部处理完毕，即全部写到网络I/O中：如果为true，则说明此次请求的所有byte数据已经全部发送完毕。
                    if (!p.bb.hasRemaining()) {
                        // 已发送数据包个数
                        sentCount++;
                        // 【2、】数据包已经通过网络socket发送出去，则把此数据包p从outgoingQueue中移除，并添加到pendingQueue中
                        outgoingQueue.removeFirstOccurrence(p);
                        if (p.requestHeader != null && p.requestHeader.getType() != OpCode.ping && p.requestHeader.getType() != OpCode.auth) {
                            synchronized (pendingQueue) {
                                pendingQueue.add(p);
                            }
                        }
                    }
                }

                // 判断是否disableWrite()
                if (outgoingQueue.isEmpty()) {
                    // No more packets to send: turn off write interest flag.
                    // Will be turned on later by a later call to enableWrite(),
                    // from within ZooKeeperSaslClient (if client is configured
                    // to attempt SASL authentication), or in either doIO() or
                    // in doTransport() if not.
                    disableWrite();
                } else if (!initialized && p != null && !p.bb.hasRemaining()) {
                    // On initial connection, write the complete connect request
                    // packet, but then disable further writes until after
                    // receiving a successful connection response.  If the
                    // session is expired, then the server sends the expiration
                    // response and immediately closes its end of the socket.  If
                    // the client is simultaneously writing on its end, then the
                    // TCP stack may choose to abort with RST, in which case the
                    // client would never receive the session expired event.  See
                    // http://docs.oracle.com/javase/6/docs/technotes/guides/net/articles/connection_release.html
                    disableWrite();
                } else {
                    // Just in case
                    enableWrite();
                }
            }
        }
    }

    /**
     * 从outgoingQueue中找到一个‘可发送的’数据包
     *
     * @param outgoingQueue
     * @param clientTunneledAuthenticationInProgress
     * @return
     */
    private Packet findSendablePacket(LinkedList<Packet> outgoingQueue, boolean clientTunneledAuthenticationInProgress) {
        synchronized (outgoingQueue) {
            if (outgoingQueue.isEmpty()) {
                return null;
            }
            if (outgoingQueue.getFirst().bb != null // If we've already starting sending the first packet, we better finish
                    || !clientTunneledAuthenticationInProgress) {
                return outgoingQueue.getFirst();
            }

            // Since client's authentication with server is in progress,
            // send only the null-header packet queued by primeConnection().
            // This packet must be sent so that the SASL authentication process
            // can proceed, but all other packets should wait until
            // SASL authentication completes.
            // 个人理解：遍历outgoingQueue，从outgoingQueue中找到requestHeader为null的数据包，并把它移动到队列的第一个元素。
            ListIterator<Packet> iter = outgoingQueue.listIterator();
            while (iter.hasNext()) {
                Packet p = iter.next();
                if (p.requestHeader == null) {
                    // We've found the priming-packet. Move it to the beginning of the queue.
                    iter.remove();
                    outgoingQueue.add(0, p);
                    return p;
                } else {
                    // Non-priming packet: defer it until later, leaving it in the queue
                    // until authentication completes.
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("deferring non-priming packet: " + p + "until SASL authentication completes.");
                    }
                }
            }
            // no sendable packet found.
            return null;
        }
    }

    @Override
    void cleanup() {
        if (sockKey != null) {
            SocketChannel sock = (SocketChannel) sockKey.channel();
            sockKey.cancel();
            try {
                sock.socket().shutdownInput();
            } catch (IOException e) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Ignoring exception during shutdown input", e);
                }
            }
            try {
                sock.socket().shutdownOutput();
            } catch (IOException e) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Ignoring exception during shutdown output",
                              e);
                }
            }
            try {
                sock.socket().close();
            } catch (IOException e) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Ignoring exception during socket close", e);
                }
            }
            try {
                sock.close();
            } catch (IOException e) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Ignoring exception during channel close", e);
                }
            }
        }
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("SendThread interrupted during sleep, ignoring");
            }
        }
        sockKey = null;
    }

    @Override
    void close() {
        try {
            if (LOG.isTraceEnabled()) {
                LOG.trace("Doing client selector close");
            }
            selector.close();
            if (LOG.isTraceEnabled()) {
                LOG.trace("Closed client selector");
            }
        } catch (IOException e) {
            LOG.warn("Ignoring exception during selector close", e);
        }
    }

    /**
     * create a socket channel.
     * @return the created socket channel
     * @throws IOException
     */
    SocketChannel createSock() throws IOException {
        SocketChannel sock;
        sock = SocketChannel.open();
        sock.configureBlocking(false);
        sock.socket().setSoLinger(false, -1);
        sock.socket().setTcpNoDelay(true);
        return sock;
    }

    /**
     * register with the selection and connect
     * 向selector注册并连接
     * @param sock the {@link SocketChannel} 
     * @param addr the address of remote host
     * @throws IOException
     */
    void registerAndConnect(SocketChannel sock, InetSocketAddress addr) throws IOException {
        sockKey = sock.register(selector, SelectionKey.OP_CONNECT);
        boolean immediateConnect = sock.connect(addr);
        if (immediateConnect) {
            sendThread.primeConnection();
        }
    }

    @Override
    void connect(InetSocketAddress addr) throws IOException {
        SocketChannel sock = createSock();
        try {
            registerAndConnect(sock, addr);
        } catch (IOException e) {
            LOG.error("Unable to open socket to " + addr);
            sock.close();
            throw e;
        }
        initialized = false;

        /*
         * Reset incomingBuffer
         */
        lenBuffer.clear();
        incomingBuffer = lenBuffer;
    }

    /**
     * 返回已连接的远程socket IP地址
     *
     * @return 返回远程连接ip地址；如果尚未socket连接，则返回null
     */
    @Override
    SocketAddress getRemoteSocketAddress() {
        try {
            return ((SocketChannel) sockKey.channel()).socket().getRemoteSocketAddress();
        } catch (NullPointerException e) {
            return null;
        }
    }

    @Override
    SocketAddress getLocalSocketAddress() {
        try {
            return ((SocketChannel) sockKey.channel()).socket().getLocalSocketAddress();
        } catch (NullPointerException e) {
            return null;
        }
    }

    /**
     * <pre>
     *
     * Selector的wakeup()，即Selector的唤醒，为什么要有这个唤醒操作呢？
     * 那还得从Selector的选择方式 来说明，前文已经总结过Selector的选择方式有三种：select()、select(timeout)、selectNow()。
     * selectNow的选择过程是非阻塞的，与wakeup没有太大关系。
     * select(timeout)和select()的选择过程是阻塞的，其他线程如果想终止这个过程，就可以调用wakeup来唤醒。
     *
     * 原文链接：<a href="http://goldendoc.iteye.com/blog/1152079">Java NIO 的 wakeup 剖析</a>
     * NIO选择器：<a href="https://www.iteye.com/blog/goldendoc-1144310">Java NIO 选择器(Selector) 知识预备 (linux epoll)</a>
     * </pre>
     */
    @Override
    synchronized void wakeupCnxn() {
        selector.wakeup();
    }

    @Override
    void doTransport(int waitTimeOut, List<Packet> pendingQueue, LinkedList<Packet> outgoingQueue, ClientCnxn cnxn) throws IOException, InterruptedException {
        // number表示channel中存在感兴趣的事件被触发了。
        int number = selector.select(waitTimeOut);
        Set<SelectionKey> selected;
        synchronized (this) {
            selected = selector.selectedKeys();
        }
        // Everything below and until we get back to the select is
        // non blocking, so time is effectively a constant. That is
        // Why we just have to do this once, here
        updateNow();

        // ===================== 下面这些都是非阻塞的：
        for (SelectionKey k : selected) {
            // 如果已经建立socket链接 并且 channel已经完成socket连接
            if (k.isConnectable() && ((SocketChannel) k.channel()).finishConnect()) {
                // 修改最后发送和接收数据的事件
                updateLastSendAndHeard();
                // 首要的链接：即创建session request
                sendThread.primeConnection();

            } else if (k.isReadable() || k.isWritable()) {
                // 如果是读写事件：处理IO
                doIO(pendingQueue, outgoingQueue, cnxn);
            }
        }

        // 如果客户端状态是‘已连接’状态：
        if (sendThread.getZkState().isConnected()) {
            synchronized (outgoingQueue) {
                // 则判断是否能找到一个‘可发送的’packet
                if (findSendablePacket(outgoingQueue, cnxn.sendThread.clientTunneledAuthenticationInProgress()) != null) {
                    // 如果能找到，则在socket层面，关注 OP_WRITE 操作。
                    enableWrite();
                }
            }
        }
        // 处理完所有的key，之后，需要清空。
        selected.clear();
    }

    //TODO should this be synchronized?
    @Override
    void testableCloseSocket() throws IOException {
        LOG.info("testableCloseSocket() called");
        ((SocketChannel) sockKey.channel()).socket().close();
    }

    @Override
    synchronized void enableWrite() {
        int i = sockKey.interestOps();
        if ((i & SelectionKey.OP_WRITE) == 0) {
            sockKey.interestOps(i | SelectionKey.OP_WRITE);
        }
    }

    @Override
    public synchronized void disableWrite() {
        int i = sockKey.interestOps();
        if ((i & SelectionKey.OP_WRITE) != 0) {
            sockKey.interestOps(i & (~SelectionKey.OP_WRITE));
        }
    }

    /**
     * 开启读操作
     */
    synchronized private void enableRead() {
        // 获取关心的操作
        int i = sockKey.interestOps();
        // 如果关心的操作不是READ操作，则关心操作添加上READ操作
        // （比如：如果之前只关心OP_WRITE操作，OP_WRITE操作不是OP_READ：则关心操作填加上OP_READ；此时关心操作为OP_WRITE和OP_READ）
        if ((i & SelectionKey.OP_READ) == 0) {
            sockKey.interestOps(i | SelectionKey.OP_READ);
        }
    }

    @Override
    synchronized void enableReadWriteOnly() {
        sockKey.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
    }

    Selector getSelector() {
        return selector;
    }

    @Override
    void sendPacket(Packet p) throws IOException {
        SocketChannel sock = (SocketChannel) sockKey.channel();
        if (sock == null) {
            throw new IOException("Socket is null!");
        }
        p.createBB();
        ByteBuffer pbb = p.bb;
        sock.write(pbb);
    }


}
